package com.atguigu.day03;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class Flink10_Transform_RichMapFun {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //2.从端口读取数据
//        DataStreamSource<String> streamSource = env.socketTextStream("localhost", 9999);
        DataStreamSource<String> streamSource = env.readTextFile("input/word.txt");

        streamSource
                .map(new MyRichMap())
                .flatMap(
                new RichFlatMapFunction<String, String>() {
            @Override
            public void open(Configuration parameters) throws Exception {
                System.out.println("flatmap open");
            }

            @Override
            public void close() throws Exception {
                System.out.println("flatmap close");
            }

            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                System.out.println(value);
            }
        }).print();

        //TODO 3.使用map对数据做处理
//        flatMap.map(new MyRichMap()).print();

        env.execute();

    }

    public static class MyRichMap extends RichMapFunction<String,String>{
        /**
         *  在实际调用方法之前最先被调用 每个并行度调用一次  可以将初始化代码写在这个方法中 比如创建链接
         * @param parameters The configuration containing the parameters attached to the contract.
         * @throws Exception
         */
        @Override
        public void open(Configuration parameters) throws Exception {
            System.out.println("mapopen");
        }

        /**
         *  在实际调用方法之前最后被调用 每个并行度调用一次  适合做一些清理工作 比如关闭链接
         * @throws Exception
         */
        @Override
        public void close() throws Exception {
            System.out.println("mapclose");
        }


        @Override
        public String map(String value) throws Exception {
           /* RuntimeContext runtimeContext = getRuntimeContext();
            System.out.println(runtimeContext.getJobId());
            System.out.println(runtimeContext.getTaskName());
            System.out.println(runtimeContext.getIndexOfThisSubtask());
            System.out.println(runtimeContext.getTaskNameWithSubtasks());*/

            return value;
        }
    }
}
